package com.hex.ds.code.detection.publish.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.io.file.FileNameUtil;
import cn.hutool.core.util.StrUtil;
//import liquibase.util.file.FilenameUtils;
import cn.hutool.db.ds.simple.SimpleDataSource;
import com.hex.ds.code.detection.common.dao.ClusterSparingConfMapper;
import com.hex.ds.code.detection.common.entity.ClusterSparingConf;
import org.apache.commons.lang3.StringUtils;
import com.hex.ds.code.detection.config.dao.ConfigPathGitMapper;
import com.hex.ds.code.detection.config.model.ConfigPropertyJdbcconfig;
import com.hex.ds.code.detection.hive.service.HiveBaseSqlVisitor;
import com.hex.ds.code.detection.oracle.service.OracleBasicSqlVisitor;
import com.hex.ds.code.detection.pack.service.IScanScriptFileService;
import com.hex.ds.code.detection.pack.service.IScriptPackTaskService;
import com.hex.ds.code.detection.publish.entity.ScriptPublishListEntity;
import com.hex.ds.code.detection.publish.entity.ScriptPublishResultEntity;
import com.hex.ds.code.detection.publish.entity.ScriptPublishTaskEntity;
import com.hex.ds.code.detection.publish.service.ConfigService;
import com.hex.ds.code.detection.publish.service.IPublishTaskService;
import com.hex.ds.code.detection.publish.service.IScriptPublishListService;
import com.hex.ds.code.detection.publish.service.IScriptPublishResultService;
import com.hex.ds.code.detection.publish.service.IScriptPublishTaskService;
import com.hex.ds.code.detection.common.util.DataSourceUtil;
import com.hex.ds.code.detection.config.service.ConfigPropertyJdbcconfigService;
import com.hex.ds.code.detection.config.service.ConfigPropertyParamService;
import com.hex.ds.code.detection.config.service.ConfigPropertyPublishuserService;
import com.hex.ds.code.detection.config.service.ConfigPropertyServerService;
import com.hex.ds.code.detection.transfer.service.ScpTransferService;
import com.jcraft.jsch.Session;
//import liquibase.util.file.FilenameUtils;
import liquibase.util.file.FilenameUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import javax.sql.DataSource;
import java.sql.*;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = {@Lazy, @Autowired})
public class PublishTaskServiceImpl implements IPublishTaskService {

    @Value("${script.scan.periodFilePath}")
    private String periodFilePath;

//    @Value("${script.publish.configFilePath}")
//    private String configFilePath;

    @Value("${script.publish.backupFilePath}")
    private String backupFilePath;

    @Value("${local-root-info.host}")
    private String rootHost;

    @Value("${local-root-info.port}")
    private String rootPort;

    @Value("${local-root-info.username}")
    private String rootUsername;

    @Value("${local-root-info.password}")
    private String rootPassword;

    @Value("${script.publish.userLocalPath}")
    private String userLocalPath;

    @Value("${script.publish.userBackupFilePath}")
    private String userBackupPath;

    @Value("${script.pack.relativePath}")
    private String relativePath;

    private final static String USER_ROOT_DIR = "/home/{}/{}";

    private final ScpTransferService scpTransferService;

    private final IScanScriptFileService scanScriptFileService;

    private final IScriptPublishTaskService scriptPublishTaskService;

    private final IScriptPublishListService scriptPublishListService;

    private final IScriptPublishResultService scriptPublishResultService;

    private final ConfigPropertyPublishuserService configPropertyPublishuserService;

    private final ConfigPropertyServerService configPropertyServerService;

    private final ConfigPropertyJdbcconfigService configPropertyJdbcconfigService;

    private final ConfigPropertyParamService configPropertyParamService;

    private final ConfigService configService;

    private final ClusterSparingConfMapper clusterSparingConfMapper;

    private final IScriptPackTaskService scriptPackTaskService;

    private final ConfigPathGitMapper configPathGitMapper;

    /**
     * @Method execPublishTask
     * @Param
     * @param scriptPublishTask
     * @Return void
     * @Description 异步调用的方式执行发布任务
     * @Author Wang zhihao
     * @Date 2023/12/1 14:25
     * @Version V1.0
     */
    @Override
    @Async
    public void execPublishTask(ScriptPublishTaskEntity scriptPublishTask) {
        // 执行发布初始化脚本
        execPublishInit(scriptPublishTask);
        // 执行推送下载稽核脚本
        execPublishService(scriptPublishTask);
        // 执行发布周期脚本
        execPublishPeriod(scriptPublishTask);
    }

    /**
     * @Method execRestoreTask
     * @Param
     * @param scriptPublishTask
     * @Return void
     * @Description 异步调用的方式执行还原任务
     * @Author Wang zhihao
     * @Date 2023/12/1 17:44
     * @Version V1.0
     */
    @Override
    @Async
    public void execRestoreTask(ScriptPublishTaskEntity scriptPublishTask, List<ScriptPublishListEntity> scriptPublishListEntities, ClusterSparingConf clusterSparingConf) {
        for (ScriptPublishListEntity scriptPublishList : scriptPublishListEntities) {
            // 获取配置信息
            Map<String,String> scpMap = configService.getScp(scriptPublishList.getClusterName());
            try {
                // 获取发布用户信息
                Map<String,String> userInfoMap = configService.getPublishUser(scriptPublishTask.getServerIds(), scriptPublishList.getClusterName(),scriptPublishTask.getPublishUser());

                // 对指定集群做还原
                restorePeriodFromCluster(scriptPublishTask, scriptPublishList, scpMap.get("host"),scpMap.get("port"), userInfoMap, scriptPublishTask.getPublishUser());
                // 若为金桥集群需要再从其镜像集群还原周期脚本文件
                if (!(StringUtils.isEmpty(scpMap.get("mirror_host")))) {
                    restorePeriodFromCluster(scriptPublishTask, scriptPublishList, scpMap.get("mirror_host"),scpMap.get("mirror_port"), userInfoMap, scriptPublishTask.getPublishUser());
                }
                String cycleMsg = getRestoreCycleMsg(scriptPublishTask, scpMap, scpMap.get("host"),
                        scpMap.get("port"), scriptPublishList.getFilePath(), "周期脚本还原成功");
                scriptPublishListService.updateById(new ScriptPublishListEntity().setId(scriptPublishList.getId()).setState("RESTORED").setErrorMessage(cycleMsg));
                scriptPublishTaskService.modifyRestoreSuccessNum(scriptPublishTask.getId());
            } catch (Exception e) {
                log.error("还原周期脚本异常" + e.getMessage(), e);
                scriptPublishTaskService.modifyRestoreFailNum(scriptPublishTask.getId());
                String cycleMsg = getRestoreCycleMsg(scriptPublishTask,scpMap, scpMap.get("host"),
                        scpMap.get("port"), scriptPublishList.getFilePath(), e.getMessage());
                scriptPublishListService.updateById(new ScriptPublishListEntity().setId(scriptPublishList.getId()).setState("RESTOREFAIL").setErrorMessage(cycleMsg));
            }
        }
    }

    /**
     * @Method restorePeriodFromCluster
     * @Param
     * @param scriptPublishTask
     * @param scriptPublishList
     * @Return void
     * @Description 从指定集群还原周期脚本文件
     * @Author Wang zhihao
     * @Date 2023/12/1 17:47
     * @Version V1.0
     */
    private void restorePeriodFromCluster(ScriptPublishTaskEntity scriptPublishTask, ScriptPublishListEntity scriptPublishList, String host,String port, Map<String, String> userInfoMap, String user) {
        Session session = null;
        try {
            session = scpTransferService.getSession(host, Integer.parseInt(port), userInfoMap.get("username"), userInfoMap.get("password"));
            // 远端(发布目标)路径
            String cycleFilePath = scriptPublishTask.getMasterPackagePath() + periodFilePath;
            // 还原目标文件路径
            String rootAbsolutePath = StrUtil.removeAll(scriptPublishList.getFilePath(), cycleFilePath);
            restorePeriodFile(session, scriptPublishList, rootAbsolutePath, scriptPublishList.getFileName(), scriptPublishTask.getPublishUser());
        } catch (Exception e) {
            log.error("【还原周期脚本异常】" + e.getMessage(), e);
            throw new RuntimeException("【还原周期脚本异常】" + e.getMessage(), e);
        } finally {
            try {
                log.warn("正在关闭session连接");
                if (session != null) {
                    session.disconnect();
                }
                log.warn("关闭session连接成功");
            } catch (Exception e) {
                log.error("【还原周期脚本】关闭SCP连接出现异常: {}", e.getMessage(), e);
                e.printStackTrace();
            }
        }
    }

    /**
     * @Method restorePeriodFile
     * @Param
     * @param session
     * @param scriptPublishList
     * @param rootAbsolutePath
     * @param fileName
     * @Return void
     * @Description 还原周期脚本文件
     * @Author Wang zhihao
     * @Date 2023/12/1 17:48
     * @Version V1.0
     */
    private void restorePeriodFile(Session session, ScriptPublishListEntity scriptPublishList, String rootAbsolutePath, String fileName, String user) {
        try {
            if ("Y".equals(scriptPublishList.getBackupFlag())) {
                backupRestore(session, scriptPublishList, rootAbsolutePath, fileName, user);
            } else {
                scpTransferService.removeFile(session, rootAbsolutePath);
            }
        } catch (Exception e) {
            log.error("【还原周期脚本文件】异常: {}", e.getMessage(), e);
            throw new RuntimeException("【还原周期脚本文件】异常: " + e.getMessage(), e);
        }
    }

    /**
     * @Method backupRestore
     * @Param
     * @Return void
     * @Description 还原-有备份的还原
     * @Author Wang zhihao
     * @Date 2023/12/8 14:21
     * @Version V1.0
     */
    private void backupRestore(Session session, ScriptPublishListEntity scriptPublishList, String rootAbsolutePath, String fileName, String user) {
        try {
            String originalPath = backupFilePath + rootAbsolutePath.replace(".sql", "_" + scriptPublishList.getDateFlag() + ".sql");
            if (!StrUtil.equalsIgnoreCase(user, "root")) {
                originalPath = StrUtil.format(USER_ROOT_DIR, user, userBackupPath) + rootAbsolutePath.replace(".sql", "_" + scriptPublishList.getDateFlag() + ".sql");
            }
            if (scpTransferService.isExistsPath(session, originalPath)) {
                // 删除发布的文件
                scpTransferService.removeFile(session, rootAbsolutePath);
                // 还原脚本文件
                String rootAbsoluteDir = StrUtil.removeAll(rootAbsolutePath, fileName);
                scpTransferService.moveFile(session, originalPath, rootAbsoluteDir, fileName);
            } else {
                throw new RuntimeException("备份文件" + originalPath + "丢失，还原异常终止");
            }
        } catch (Exception e) {
            log.error("【存在备份的文件还原】异常: {}", e.getMessage(), e);
            throw new RuntimeException("【存在备份的文件还原】异常: " + e.getMessage(), e);
        }
    }

    /**
     * @Method execPublishInit
     * @Param
     * @param scriptPublishTask
     * @Return void
     * @Description 执行发布初始化脚本
     * @Author Wang zhihao
     * @Date 2023/11/30 19:59
     * @Version V1.0
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void execPublishInit(ScriptPublishTaskEntity scriptPublishTask) {
        AtomicReference<String> scanFileName = new AtomicReference<>("");
        // 获取所有要发布的初始化脚本
        List<ScriptPublishListEntity> scriptPublishListEntities = scriptPublishListService.queryListNoPage(
                new ScriptPublishListEntity().setScriptPublishTaskId(scriptPublishTask.getId()).setPublishType("init"));
        // 对发布的初始化脚本做排序【顺序优先级排序为：DROP, CREATE, INIT, ALTER】
        scriptPublishListEntities.sort(getFileNameComparator());
        // 获取连接与获取特殊参数映射信息
        DataSource impalaDataSource = dynamicImpalaDataSource(scriptPublishTask.getPublishClusterNames());
        Map<String,String> impalaConnInfo = configService.getJdbc("impala",scriptPublishTask.getPublishClusterNames());
        Map<String,String> paramsMap = configService.getParam(scriptPublishTask.getPublishClusterNames());
        // 遍历去做表的提前备份操作
//        backupInitTable(scriptPublishListEntities, impalaDataSource, paramsMap, scriptPublishTask.getDateFlag());
        // 遍历去执行发布任务
        scriptPublishListEntities.forEach(scriptPublishList -> {
            try {
                scanFileName.set(scriptPublishList.getFileName());
                // 获取SQL中有改动的表(需要备份的表)
                List<String> modifyTableName = getModifyTable(scriptPublishList.getFilePath(), paramsMap);
                // 备份表数量检查(若已存在三个备份表，则删除掉时间距离最久的备份表及信息)
//                checkBackupTable(scriptPublishTask.getId(), modifyTableName, scriptPublishList.getClusterId(), impalaDataSource, "init", scriptPublishList.getFileName());
                // 备份表
                String backupTable = backupTable(modifyTableName, impalaDataSource, scriptPublishList.getDateFlag());
                // 执行sql
                executeInitScript(scriptPublishList.getFilePath(), impalaDataSource, paramsMap);
                // 获取执行日志
                String msg = getMsg(scriptPublishTask.getPublishUser(),
                        impalaConnInfo.get("url"),
                        impalaConnInfo.get("username"),
                        "初始化脚本发布成功！" + (StrUtil.isBlank(backupTable) ? "" : "该次执行存在创建备份表：" + backupTable),
                        getExecSql(scriptPublishList.getFilePath(), paramsMap));
                // 更新状态
                scriptPublishListService.updateById(new ScriptPublishListEntity().setId(scriptPublishList.getId()).setState("PUBLISHED").setErrorMessage(msg));
                scriptPublishTaskService.modifyPublishSuccessNum(scriptPublishTask.getId());
            } catch (Exception e) {
                log.error("脚本" + scanFileName + "发布初始化脚本异常 {}", e.getMessage(), e);
                scriptPublishTaskService.modifyPublishFailNum(scriptPublishTask.getId());
                // 获取执行日志
                String msg = getMsg(scriptPublishTask.getPublishUser(),
                        impalaConnInfo.get("url"),
                        impalaConnInfo.get("username"),
                        e.getMessage(),
                        getExecSql(scriptPublishList.getFilePath(), paramsMap));
                scriptPublishListService.updateById(new ScriptPublishListEntity().setId(scriptPublishList.getId()).setState("PUBLISHFAIL").setErrorMessage(msg));
            }
        });
    }

    private void sortInitScriptBySuffix(List<ScriptPublishListEntity> scriptPublishListEntities) {
        scriptPublishListEntities.sort(getFileNameComparator());
    }

    /**
     * @Method getFileNameComparator
     * @Param
     * @Return java.util.Comparator<com.hex.ds.code.detection.script.publish.entity.ScriptPublishListEntity>
     * @Description 比较器
     * @Author Wang zhihao
     * @Date 2024/4/8 19:20
     * @Version V1.0
     */
    private Comparator<ScriptPublishListEntity> getFileNameComparator() {
        return Comparator.comparing(this::getFileNamePriority);
    }

    // 规则设定
    private int getFileNamePriority(ScriptPublishListEntity scriptPublishListEntity) {

        String fileNameWithoutExtension = FilenameUtils.getBaseName(scriptPublishListEntity.getFileName());

        String[] wordsInName = fileNameWithoutExtension.split("_");
        String lastWord = wordsInName[wordsInName.length - 1];

        Map<String, Integer> wordPriorityMap = new HashMap<>();
        wordPriorityMap.put("drop", 1);
        wordPriorityMap.put("create", 2);
        wordPriorityMap.put("init", 3);
        wordPriorityMap.put("alter", 4);

        return wordPriorityMap.getOrDefault(lastWord, Integer.MAX_VALUE);
    }

    /**
     * @Method getMsg
     * @Param publishUser 发布用户
     * @param jdbcUrl 数据库链接
     * @param jdbcName 数据库用户名
     * @param msg 发布情况
     * @param sql 执行sql
     * @Return String
     * @Description 拼接初始化/推送下载稽核日志
     * @Author gxd
     * @Date 2024/01/11 15:50
     * @Version V1.0
     */
    private String getMsg(String publishUser,String jdbcUrl,String jdbcName, String msg,String sql){
        StringBuilder msgBuillder = new StringBuilder();
        msgBuillder.append("【发布用户】：").append("\n").append(publishUser).append("\n\n")
                .append("【数据库链接】：").append("\n").append(jdbcUrl).append("\n\n")
                .append("【数据库用户名】：").append("\n").append(jdbcName).append("\n\n")
                .append("【发布情况】：").append("\n").append(msg).append("\n\n")
                .append("【执行sql】：").append("\n").append(sql);
        return msgBuillder.toString();
    }
    /**
     * @Method getCycleMsg
     * @Param publishUser 发布用户
     * @param host 服务器Host
     * @param port 服务器Port
     * @param path 存放路径
     * @param msg 发布情况
     * @Return String
     * @Description 拼接周期脚本发布日志
     * @Author gxd
     * @Date 2024/01/11 15:50
     * @Version V1.0
     */
    private String getCycleMsg(ScriptPublishListEntity scriptPublishList, String publishUser, Map<String, String> scpConnInfo, String host,String port, String path, String msg){
        // 源文件路径
        String sourcePath = scriptPublishList.getFilePath();
        if (!StrUtil.equalsIgnoreCase(publishUser, "root")) {
            sourcePath = sourcePath.replace(relativePath + "/pack",  StrUtil.format(USER_ROOT_DIR, publishUser, userLocalPath));
        }
        // 是否存在镜像服务器
        boolean haveMirror = !(StringUtils.isEmpty(scpConnInfo.get("mirror_host")));
        StringBuilder msgBuilder = new StringBuilder();
        msgBuilder.append("【发布用户】：").append("\n").append(publishUser).append("\n\n")
                .append("【服务器Host】：").append("\n").append(host).append("\n\n")
                .append("【服务器Port】：").append("\n").append(port).append("\n\n")
                .append( haveMirror ? "【镜像服务器Host】：\n" + scpConnInfo.get("mirror_host") + "\n\n" : "")
                .append( haveMirror ? "【镜像服务器Port】：\n" + scpConnInfo.get("mirror_port") + "\n\n" : "")
                .append("【源文件存放路径】：").append("\n").append(sourcePath).append("\n\n")
                .append("【文件发布目路径】：").append("\n").append(path).append("\n\n")
                .append("【文件备份情况】：\n").append(StrUtil.isBlank(scriptPublishList.getBackupFilePath()) ? "【未发生文件备份情况】" : "【存在文件备份】：" + scriptPublishList.getBackupFilePath()).append("\n\n")
                .append("【发布情况】：").append("\n").append(msg);
        return msgBuilder.toString().replaceAll("//", "/");
    }


    private String getRestoreCycleMsg(ScriptPublishTaskEntity scriptPublishTask, Map<String,String> scpMap, String host,String port, String path, String msg){
        String cycleFilePath = scriptPublishTask.getMasterPackagePath() + periodFilePath;
        String rootAbsolutePath = path.replace("\\", "/").replace(cycleFilePath, "");
        // 是否存在镜像服务器
        boolean haveMirror = !(StringUtils.isEmpty(scpMap.get("mirror_host")));
        StringBuilder msgBuilder = new StringBuilder();
        msgBuilder.append("【还原用户】：").append("\n").append(scriptPublishTask.getPublishUser()).append("\n\n")
                .append("【服务器Host】：").append("\n").append(host).append("\n\n")
                .append("【服务器Port】：").append("\n").append(port).append("\n\n")
                .append( haveMirror ? "【镜像服务器Host】：\n" + scpMap.get("mirror_host") + "\n\n" : "")
                .append( haveMirror ? "【镜像服务器Port】：\n" + scpMap.get("mirror_port") + "\n\n" : "")
                .append("【还原目标文件路径】：").append("\n").append(rootAbsolutePath).append("\n\n")
                .append("【还原情况】：").append("\n").append(msg);
        return msgBuilder.toString();
    }

    /**
     * @Method getExecSql
     * @Param filePath
     * @param paramsMap
     * @Return String
     * @Description 获取初始化脚本sql
     * @Author gxd
     * @Date 2024/01/11 15:50
     * @Version V1.0
     */
    private String getExecSql(String filePath,Map<String, String> paramsMap){
        try {
        // 逐行读取文件去除注释掉的行
        List<String> linesInfo = FileUtil.readUtf8Lines(filePath).stream().filter(s -> !StrUtil.startWith(s, "--")).collect(Collectors.toList());
        // 去除空行
        List<String> noBlankSql = linesInfo.stream().filter(StrUtil::isNotBlank).collect(Collectors.toList());
        if (CollectionUtil.isNotEmpty(noBlankSql)) {
            // 合并SQL行为完整的SQL
            String fullSql = String.join("\n", noBlankSql);
            // 若存在需要参数替换的配置 则做参数替换
            if (CollectionUtil.isNotEmpty(paramsMap)) {
                for (Map.Entry<String, String> entry : paramsMap.entrySet()) {
                    fullSql = fullSql.replace(entry.getKey(), entry.getValue());
                }
            }
            return fullSql;
        }else {
            return "该文件不存在可执行的sql内容。";
        }
        }catch (Exception e){
            return "获取sql失败！";
        }
    }
    /**
     * @Method execPublishService
     * @Param
     * @param scriptPublishTask
     * @Return void
     * @Description 执行推送下载稽核脚本
     * @Author Wang zhihao
     * @Date 2023/11/30 19:59
     * @Version V1.0
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void execPublishService(ScriptPublishTaskEntity scriptPublishTask) {
        AtomicReference<String> scanFileName = new AtomicReference<>("");
        List<ScriptPublishListEntity> scriptPublishListEntities = scriptPublishListService.queryListNoPage(
                new ScriptPublishListEntity().setScriptPublishTaskId(scriptPublishTask.getId()).setPublishType("service"));
        // 遍历去做表的提前备份操作
//        backupServiceTable(scriptPublishListEntities);
        scriptPublishListEntities.forEach(scriptPublishList -> {
            Map<String,String> oracleConnectInfo = configService.getJdbc(scriptPublishList.getFilePath().split("/")[scriptPublishList.getFilePath().split("/").length-2].toLowerCase(),scriptPublishList.getClusterName());
            try {
                scanFileName.set(scriptPublishList.getFileName());
                DataSource dataSource = dynamicJdbcDataSource(scriptPublishList.getClusterName(), scriptPublishList.getFilePath());
                // 获取Oracle脚本中存在数据变动的表
                List<String> modifyTableName = getOracleModifyTable(scriptPublishList.getFilePath());
                // 备份表数量检查(若已存在三个备份表，则删除掉时间距离最久的备份表及信息)
//                checkBackupTable(scriptPublishTask.getId(), modifyTableName, scriptPublishList.getClusterId(), dataSource, "service", scanFileName.toString());
                // 备份表
                String backupTableName = backupOracleTable(modifyTableName, dataSource, scriptPublishList.getDateFlag());
                // 执行SQL
                executeServiceScript(scriptPublishList.getFilePath(), dataSource);
                // 获取执行日志
                String msg = getMsg(scriptPublishTask.getPublishUser(),
                        oracleConnectInfo.get("url"),
                        oracleConnectInfo.get("username"),
                        "发布推送下载稽核脚本成功！" + (StrUtil.isBlank(backupTableName) ? "" : "该次执行存在创建备份表：" + backupTableName),
                        String.join("\n", getSplitSql(String.join("\n", FileUtil.readUtf8Lines(scriptPublishList.getFilePath())))));
                // 更新状态
                scriptPublishListService.updateById(new ScriptPublishListEntity().setId(scriptPublishList.getId()).setState("PUBLISHED").setErrorMessage(msg));
                scriptPublishTaskService.modifyPublishSuccessNum(scriptPublishTask.getId());
            } catch (Exception e) {
                log.error("脚本" + scanFileName + "发布推送下载稽核脚本异常 {}", e.getMessage(), e);
                scriptPublishTaskService.modifyPublishFailNum(scriptPublishTask.getId());
                // 获取执行日志
                String msg = getMsg(scriptPublishTask.getPublishUser(),
                        oracleConnectInfo.get("url"),
                        oracleConnectInfo.get("username"),
                        e.getMessage(),
                        String.join("\n", getSplitSql(String.join("\n", FileUtil.readUtf8Lines(scriptPublishList.getFilePath())))));
                scriptPublishListService.updateById(new ScriptPublishListEntity().setId(scriptPublishList.getId()).setState("PUBLISHFAIL").setErrorMessage(msg));
            }
        });
    }

    /**
     * @Method execPublishPeriod
     * @Param
     * @Return void
     * @Description 执行发布周期脚本
     * @Author Wang zhihao
     * @Date 2023/11/30 19:59
     * @Version V1.0
     */
    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
    public void execPublishPeriod(ScriptPublishTaskEntity scriptPublishTask) {
        // 若发布脚本用户非root，调整文件位置
        adjustFilePathByUser(scriptPublishTask.getPublishUser(), StrUtil.format("{}/pack/{}", relativePath, scriptPublishTask.getPackTaskId()));
        List<ScriptPublishListEntity> scriptPublishListEntities = scriptPublishListService.queryListNoPage(
                new ScriptPublishListEntity().setScriptPublishTaskId(scriptPublishTask.getId()).setPublishType("period"));
        scriptPublishListEntities.forEach(scriptPublishList -> {
            // 获取配置信息
            Map<String,String> scpConnInfo = configService.getScp(scriptPublishList.getClusterName());
            //获取periodFilePath
            // 获取发布目标文件路径
            String cycleFilePath = scriptPublishTask.getMasterPackagePath() + periodFilePath;
            String rootAbsolutePath = scriptPublishList.getFilePath().replace("\\","/").replace(cycleFilePath, "");
            try {
                // 获取用户信息
                Map<String,String> userInfoMap = configService.getPublishUser(scriptPublishTask.getServerIds(), scriptPublishList.getClusterName(),scriptPublishTask.getPublishUser());
                // 执行指定集群的发布
                execPublishPeriodCluster(rootAbsolutePath, scriptPublishList, scpConnInfo, userInfoMap, scriptPublishTask.getPublishUser());
                // 若存在镜像服务器，则对镜像服务器也进行发布
                if (!(StringUtils.isEmpty(scpConnInfo.get("mirror_host")))) {
                    Map<String,String> map = new HashMap<>();
                    map.put("host",scpConnInfo.get("mirror_host"));
                    map.put("port",scpConnInfo.get("mirror_port"));
                    execPublishPeriodCluster(rootAbsolutePath, scriptPublishList, map, userInfoMap, scriptPublishTask.getPublishUser());
                }
                scriptPublishTaskService.modifyPublishSuccessNum(scriptPublishTask.getId());
                // 拼接周期脚本发布日志
                String cycleMsg = getCycleMsg(scriptPublishList, scriptPublishTask.getPublishUser(), scpConnInfo, scpConnInfo.get("host"),
                        scpConnInfo.get("port"), rootAbsolutePath,"周期脚本发布成功！");
                scriptPublishListService.updateById(new ScriptPublishListEntity().setId(scriptPublishList.getId()).setErrorMessage(cycleMsg));

            } catch (Exception e) {
                log.error("脚本" + scriptPublishList.getFileName() + "发布周期脚本异常 {}", e.getMessage(), e);
                scriptPublishTaskService.modifyPublishFailNum(scriptPublishTask.getId());
                // 拼接周期脚本发布日志
                String cycleMsg = getCycleMsg(scriptPublishList, scriptPublishTask.getPublishUser(), scpConnInfo, scpConnInfo.get("host"),
                        scpConnInfo.get("port"), rootAbsolutePath,e.getMessage());
                scriptPublishListService.updateById(new ScriptPublishListEntity().setId(scriptPublishList.getId()).setState("PUBLISHFAIL").setErrorMessage(cycleMsg));
            }
        });
    }

    /**
     * @Method adjustFilePathByUser
     * @Param
     * @param
     * @Return void
     * @Description 若发布用户非root, 调整脚本存放路径
     * @Author Wang zhihao
     * @Date 2024/1/25 19:40
     * @Version V1.0
     */
    private void adjustFilePathByUser(String publishUser, String savePath) {
        if (!StrUtil.equalsIgnoreCase(publishUser, "root")) {
            Session session = null;
            try {
                session = scpTransferService.getSession(rootHost, Integer.parseInt(rootPort), rootUsername, rootPassword);
                scpTransferService.copyFileLocal(session, savePath, StrUtil.format(USER_ROOT_DIR, publishUser, userLocalPath));
                scpTransferService.chmodDir(session, StrUtil.format(USER_ROOT_DIR, publishUser, userLocalPath), publishUser);
            } catch (Exception e) {
                log.error("【发布周期脚本】调整脚本存放路径出现异常: {}", e.getMessage(), e);
                throw new RuntimeException("【发布周期脚本】调整脚本存放路径出现异常" + e.getMessage(), e);
            } finally {
                try {
                    session.disconnect();
                } catch (Exception e) {
                    log.error("【发布周期脚本】调整脚本存放路径关闭SCP连接出现异常: {}", e.getMessage(), e);
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * @Method execPublishPeriodCluster
     * @Param
     * @param scriptPublishList
     * @param scpConnInfo
     * @Return void
     * @Description 发布周期脚本文件到集群
     * @Author Wang zhihao
     * @Date 2023/11/30 20:00
     * @Version V1.0
     */
    private void execPublishPeriodCluster(String rootAbsolutePath, ScriptPublishListEntity scriptPublishList, Map<String, String> scpConnInfo, Map<String, String> userInfoMap, String user) {
        Session session = null;
        try {
            // 源文件路径
            String sourcePath = scriptPublishList.getFilePath();
            if (!StrUtil.equalsIgnoreCase(user, "root")) {
                sourcePath = sourcePath.replace(relativePath + "/pack",  StrUtil.format(USER_ROOT_DIR, user, userLocalPath));
            }
            session = scpTransferService.getSession(scpConnInfo.get("host"), Integer.parseInt(scpConnInfo.get("port")), userInfoMap.get("username"), userInfoMap.get("password"));
            // 发布文件
            publishPeriodFile(rootAbsolutePath, sourcePath, session, scriptPublishList, user);
        } catch (Exception e) {
            log.error("【发布周期脚本】出现异常: {}", e.getMessage(), e);
            throw new RuntimeException("【发布周期脚本】出现异常" + e.getMessage(), e);
        } finally {
            try {
                log.warn("【发布周期脚本】关闭SCP连接开始");
                if (session != null) {
                    session.disconnect();
                }
                log.warn("【发布周期脚本】关闭SCP连接完成");
            } catch (Exception e) {
                log.error("【发布周期脚本】关闭SCP连接出现异常: {}", e.getMessage(), e);
                e.printStackTrace();
            }
        }
    }

    /**
     * @Method executeInitScript
     * @Param
     * @param filePath
     * @param dataSource
     * @Return void
     * @Description 执行初始化脚本文件
     * @Author Wang zhihao
     * @Date 2023/11/14 9:50
     * @Version V1.0
     */
    private void executeInitScript(String filePath, DataSource dataSource, Map<String, String> paramsMap) {
        try {
            // 逐行读取文件去除注释掉的行
            List<String> linesInfo = FileUtil.readUtf8Lines(filePath).stream().filter(s -> !StrUtil.startWith(s, "--")).collect(Collectors.toList());
            // 去除空行
            List<String> noBlankSql = linesInfo.stream().filter(StrUtil::isNotBlank).collect(Collectors.toList());
            if (CollectionUtil.isNotEmpty(noBlankSql)) {
                // 合并SQL行为完整的SQL
                String fullSql = String.join("\n", noBlankSql);
                // 若存在需要参数替换的配置 则做参数替换
                if (CollectionUtil.isNotEmpty(paramsMap)) {
                    for (Map.Entry<String, String> entry : paramsMap.entrySet()) {
                        fullSql = fullSql.replace(entry.getKey(), entry.getValue());
                    }
                }
                // 整合逐行读取的sql语句并以;分割为多段SQL
                List<String> allSqlList = Arrays.stream(fullSql.split(";")).collect(Collectors.toList());
                executeSql(dataSource, allSqlList);
            }
        } catch (Exception e) {
            log.error("【执行初始化脚本文件】异常: {}", e.getMessage(), e);
            throw new RuntimeException("【执行初始化脚本文件】异常: " + e.getMessage(), e);
        }
    }

    /**
     * @Method executeServiceScript
     * @Param
     * @param filePath
     * @param dataSource
     * @Return void
     * @Description 执行推送下载稽核脚本的SQL
     * @Author Wang zhihao
     * @Date 2023/11/14 17:52
     * @Version V1.0
     */
    private void executeServiceScript(String filePath, DataSource dataSource) {
        try {
            List<String> allSql = getSplitSql(String.join("\n", FileUtil.readUtf8Lines(filePath)));
            executeSql(dataSource, allSql);
        } catch (Exception e) {
            log.error("【执行推送下载稽核脚本的SQL】异常: {}", e.getMessage(), e);
            throw new RuntimeException("【执行推送下载稽核脚本的SQL】异常: " + e.getMessage(), e);
        }
    }

    /**
     * @Method getSplitSql
     * @Param
     * @param fullSql
     * @Return java.util.List<java.lang.String>
     * @Description 借用Anltr实现对Oracle sql的切分
     * @Author Wang zhihao
     * @Date 2023/12/21 16:49
     * @Version V1.0
     */
    private List<String> getSplitSql(String fullSql) {
        try {
            OracleBasicSqlVisitor oracleBasicSqlVisitor = new OracleBasicSqlVisitor();
            oracleBasicSqlVisitor.visitRoot(fullSql);
            return oracleBasicSqlVisitor.getSplitSql();
        } catch (Exception e) {
            throw new RuntimeException("【执行推送下载稽核脚本的SQL】切分SQL异常");
        }
    }

    /**
     * @Method getModifyTable
     * @Param
     * @param filePath
     * @Return java.util.List<java.lang.String>
     * @Description 获取有改动的表
     * @Author Wang zhihao
     * @Date 2023/11/9 19:29
     * @Version V1.0
     */
    private List<String> getModifyTable(String filePath, Map<String, String> paramsMap) {
        try {
            List<String> allDatabaseName = new ArrayList<>();
            filePath = filePath.replace("\\","/");
            // 逐块获取去除注释掉的SQL块
            List<String> linesInfo = scanScriptFileService.getSqlByReadFile(filePath);
            // 做参数替换并去掉空行
            List<String> replaceSql = new ArrayList<>();
            for (String line : linesInfo) {
                for (Map.Entry<String, String> entry : paramsMap.entrySet()) {
                    line = line.replace(entry.getKey(), entry.getValue());
                }
                replaceSql.add(line);
            }
            // 获取动表数据操作的所有表名
            for (String sql : replaceSql) {
                HiveBaseSqlVisitor hiveBaseSqlVisitor = new HiveBaseSqlVisitor();
                hiveBaseSqlVisitor.visitRoot(sql);
                List<String> tableName = hiveBaseSqlVisitor.getDatabaseNameList();
                allDatabaseName.addAll(tableName);
            }
            // 返回去重之后的list
            return allDatabaseName.stream().distinct().collect(Collectors.toList());
        } catch (Exception e) {
            log.error("【获取有改动的表】异常: {}", e.getMessage(), e);
            throw new RuntimeException("【获取有改动的表】异常: " + e.getMessage(), e);
        }
    }

    /**
     * @Method getOracleModifyTable
     * @Param
     * @param filePath
     * @Return java.util.List<java.lang.String>
     * @Description 获取Oracle脚本中存在数据变动的表
     * @Author Wang zhihao
     * @Date 2023/11/14 14:49
     * @Version V1.0
     */
    private List<String> getOracleModifyTable(String filePath) {
        try {
            filePath = filePath.replace("\\","/");
            // 获取完整的SQL
            String fullSql = String.join("\n", FileUtil.readUtf8Lines(filePath));
            OracleBasicSqlVisitor oracleBasicSqlVisitor = new OracleBasicSqlVisitor();
            oracleBasicSqlVisitor.visitRoot(fullSql);
            List<String> tableName = oracleBasicSqlVisitor.getDatabaseNameList();
            // 返回去重之后的list
            return tableName.stream().distinct().collect(Collectors.toList());
        } catch (Exception e) {
            log.error("【获取Oracle脚本中存在数据变动的表】异常: {}", e.getMessage(), e);
            throw new RuntimeException("【获取Oracle脚本中存在数据变动的表】异常: " + e.getMessage(), e);
        }
    }

    /**
     * @Method backupTable
     * @Param
     * @param allTableName
     * @param dataSource
     * @Return void
     * @Description 备份表
     * @Author Wang zhihao
     * @Date 2023/11/9 18:50
     * @Version V1.0
     */
    private String backupTable(List<String> allTableName, DataSource dataSource, String dateFlag) {
        String backupTable = "";
        try {
            for (String tableName : allTableName) {
                // 校验表是否已存在需要备份
                String[] allTablePiece = tableName.split("\\.");
                String databaseNamePiece = allTablePiece.length > 0 ? allTablePiece[0] : "";
                String tableNamePiece = allTablePiece.length > 1 ? allTablePiece[1] : "";
                String backupTableName = tableName + "_" + dateFlag;
                if (isImpalaTableExist(dataSource, databaseNamePiece, tableNamePiece) && !isImpalaTableExist(dataSource, databaseNamePiece, tableNamePiece + "_" + dateFlag)) {
                    // JDBC的方式获取表信息
                    String tableInfo = getTableInfo(tableName, dataSource);
                    // 根据正则的方式获取分区字段信息
                    String partitionInfo = getPartitionByReg(tableInfo);
                    // 添加发布结果信息  --20240411 由于不需要还原了 废除
//                    addPublishResultInfo(listId, tableName, partitionInfo);
                    String backupTableOrPartition = backupTableName;
                    if (StrUtil.isNotBlank(partitionInfo)) {
                        backupTableOrPartition += StrUtil.format(" PARTITION({})", partitionInfo);
                    }
                    List<String> sqlList = new ArrayList<>();
                    // 建备份表语句
                    sqlList.add(StrUtil.format("CREATE TABLE {} LIKE {}", backupTableName, tableName));
                    // 备份数据语句
                    sqlList.add(StrUtil.format("INSERT OVERWRITE {} SELECT * FROM {}", backupTableOrPartition, tableName));
                    executeSql(dataSource, sqlList);
                    backupTable = backupTable + tableNamePiece + "_" + dateFlag + "  ";
                } else {
                    log.warn("【备份表初始化表】告警：表{}不存在，不做备份操作", tableName);
                }
            }
        } catch (Exception e) {
            log.error("【备份表初始化表】异常: {}", e.getMessage(), e);
            throw new RuntimeException("【备份表初始化表】异常: " + e.getMessage(), e);
        }
        return backupTable;
    }

    /**
     * @Method backupOracleTable
     * @Param
     * @param allTableName
     * @param dataSource
     * @param dateFlag
     * @Return void
     * @Description 备份Oracle表
     * @Author Wang zhihao
     * @Date 2023/11/14 15:55
     * @Version V1.0
     */
    private String backupOracleTable(List<String> allTableName, DataSource dataSource, String dateFlag) {
        String backupTable = "";
        try {
            for (String tableName : allTableName) {
                // 若tableName中存在. 则取 .后的字符串
                String[] allTablePiece = tableName.split("\\.");
                tableName = allTablePiece.length > 1 ? allTablePiece[1] : allTablePiece[0];
                String backupTableName = tableName + "_" + dateFlag;
                // 校验表是否已存在
                if (isOracleTableExist(dataSource, tableName) && !isOracleTableExist(dataSource, backupTableName)) {
                    // 添加发布结果信息
//                    addPublishResultInfo(listId, tableName, null);
                    List<String> sqlList = new ArrayList<>();
                    // 备份表结构与数据
                    sqlList.add(StrUtil.format("CREATE TABLE {} AS SELECT * FROM {}", backupTableName, tableName));
                    executeSql(dataSource, sqlList);
                    backupTable = backupTable + backupTableName + "  ";
                } else {
                    log.warn("【备份Oracle表】告警: 表{}不存在", tableName);
                }
            }
        } catch (Exception e) {
            log.error("【备份Oracle表】异常: {}", e.getMessage(), e);
            throw new RuntimeException("【备份Oracle表】异常: " + e.getMessage(), e);
        }
        return backupTable;
    }

    /**
     * @Method checkBackupTable
     * @Param
     * @param taskId
     * @param tableNames
     * @param clusterId
     * @param dataSource
     * @Return void
     * @Description 检查备份表是否超过3张 是则删除时间最久的
     * @Author Wang zhihao
     * @Date 2023/11/13 20:03
     * @Version V1.0
     */
    private void checkBackupTable(String taskId, List<String> tableNames, String clusterId, DataSource dataSource, String type, String scanFileName) {
        try {
            // 获取指定文件的所有成功备份并根据时间排序
            List<ScriptPublishListEntity> scriptPublishLists = scriptPublishListService.queryListNoPage(new ScriptPublishListEntity()
                    .setScriptPublishTaskId(taskId).setClusterId(clusterId).setPublishType(type).setFileName(scanFileName).setState("PUBLISHED"));
            List<ScriptPublishListEntity> scriptPublishSortLists = ListUtil.sortByProperty(scriptPublishLists, "backupTime");
            if (scriptPublishSortLists.size() > 2) {
                // 获取时间最久的备份信息
                ScriptPublishListEntity scriptPublishList = scriptPublishSortLists.get(0);
                // 删除表
                dropBackupTable(tableNames, dataSource, scriptPublishList.getDateFlag());
                // 删除备份信息
                scriptPublishListService.removeById(scriptPublishList.getId());
            }
        } catch (Exception e) {
            log.error("【检查备份表是否超过3张,删除时间最久的表】异常: {}", e.getMessage(), e);
            throw new RuntimeException("【检查备份表是否超过3张,删除时间最久的表】异常: " + e.getMessage(), e);
        }
    }

    /**
     * @Method dropBackupTable
     * @Param
     * @param tableNames
     * @param dataSource
     * @param timeFlag
     * @Return void
     * @Description 删除备份表
     * @Author Wang zhihao
     * @Date 2023/11/15 15:55
     * @Version V1.0
     */
    private void dropBackupTable(List<String> tableNames, DataSource dataSource, String timeFlag) {
        try {
            for (String tableName : tableNames) {
                List<String> sqlList = new ArrayList<>();
                sqlList.add(StrUtil.format("DROP TABLE {}", tableName + "_" + timeFlag));
                executeSql(dataSource, sqlList);
            }
        } catch (Exception e) {
            log.error("【删除备份表】所删除的表{}不存在", tableNames + "_" + timeFlag);
            throw new RuntimeException("【删除备份表】所删除的表" + tableNames + "_" + timeFlag + "不存在");
        }
    }

    /**
     * @Method addPublishResultInfo
     * @Param
     * @param ListId
     * @param tableName
     * @param partitionInfo
     * @Return java.lang.String
     * @Description 添加发布结果信息
     * @Author Wang zhihao
     * @Date 2023/11/13 15:53
     * @Version V1.0
     */
    private String addPublishResultInfo(String ListId, String tableName, String partitionInfo) {
        ScriptPublishResultEntity scriptPublishResult = new ScriptPublishResultEntity()
                .setScriptPublishListId(ListId)
                .setOriginalPath(tableName)
                .setPartition(partitionInfo);
        scriptPublishResultService.save(scriptPublishResult);
        return scriptPublishResult.getId();
    }

    /**
     * @Method getTableInfo
     * @Param
     * @param tableName
     * @param dataSource
     * @Return java.lang.String
     * @Description 根据表名获取表的建表信息
     * @Author Wang zhihao
     * @Date 2023/11/9 18:50
     * @Version V1.0
     */
    private String getTableInfo(String tableName, DataSource dataSource) {
        Connection connection = null;
        Statement statement = null;
        ResultSet resultSet = null;
        String result = "";
        try {
            connection = dataSource.getConnection();
            statement = connection.createStatement();
            resultSet = statement.executeQuery(StrUtil.format("SHOW CREATE TABLE {}", tableName));
            while (resultSet.next()) {
                ResultSetMetaData metaData = resultSet.getMetaData();
                result = resultSet.getString(1);
            }
            return result;
        } catch (Exception e) {
            e.printStackTrace();
            log.error("【执行SQL语句】异常 {}", e.getMessage(), e);
            throw new RuntimeException("【执行SQL语句】异常: " + e.getMessage(), e);
        } finally {
            try {
                resultSet.close();
                statement.close();
                connection.close();
            } catch (Exception e) {
                e.printStackTrace();
                log.error("关闭JDBC连接异常 {}", e.getMessage(), e);
            }
        }
    }

    /**
     * @Method getPartitionByReg
     * @Param
     * @param tableInfo
     * @Return java.lang.String
     * @Description 根据正则的方式获取分区字段信息
     * @Author Wang zhihao
     * @Date 2023/11/9 17:14
     * @Version V1.0
     */
    private String getPartitionByReg(String tableInfo) {
        // 先将DECIMAL(x,x)的转为DECIMAL防止影响后续的提取分区字段
        String newTableInfo = tableInfo.replaceAll("DECIMAL\\(\\w+,\\w+\\)", "DECIMAL");
        // 正则的方式提取分区字段
        Pattern pattern = Pattern.compile("PARTITIONED BY \\((.*?)\\)", Pattern.DOTALL);
        Matcher matcher = pattern.matcher(newTableInfo);
        if (matcher.find()) {
            String partitionInfo = matcher.group(1).replaceAll("\n", "");
            List<String> partitionFields = Arrays.stream(partitionInfo.split(",")).map(s -> s.trim().split(" ")[0]).collect(Collectors.toList());
            return String.join(",", partitionFields);
        } else {
            return "";
        }
    }

    /**
     * @Method dynamicJdbcDataSource
     * @Param
     * @param filePath
     * @Return javax.sql.DataSource
     * @Description 动态根据目录获取对应的Oracle连接
     * @Author Wang zhihao
     * @Date 2023/10/23 18:58
     * @Version V1.0
     */
    private DataSource dynamicJdbcDataSource(String clusterName, String filePath) {
        try {
            String dbName = filePath.split("/")[filePath.split("/").length-2];
            String clusterId = clusterSparingConfMapper.getId(clusterName);
            ConfigPropertyJdbcconfig queryOneByInfo = configPropertyJdbcconfigService.queryOneByInfo(
                    new ConfigPropertyJdbcconfig().setClusterId(clusterId).setJdbcName(dbName));
            return new SimpleDataSource(queryOneByInfo.getUrl(), queryOneByInfo.getUsername(), queryOneByInfo.getPassword(), queryOneByInfo.getDriverclassname());
        } catch (Exception e) {
            log.error("【动态根据目录获取对应的Oracle连接】异常 " + e.getMessage(), e);
            throw new RuntimeException("【动态根据目录获取对应的Oracle连接】异常 " + e.getMessage(), e);
        }
    }

    /**
     * @Method dynamicImpalaDataSource
     * @Param
     * @param clusterName
     * @Return javax.sql.DataSource
     * @Description 动态根据目录获取对应的Impala连接
     * @Author Wang zhihao
     * @Date 2023/11/27 10:28
     * @Version V1.0
     */
    private DataSource dynamicImpalaDataSource(String clusterName) {
        try {
            List<ConfigPropertyJdbcconfig> jdbcList = configPropertyJdbcconfigService.findAll();
            String clusterId = clusterSparingConfMapper.getId(clusterName);
            for (ConfigPropertyJdbcconfig c :
                    jdbcList) {
                if (clusterId.equals(c.getClusterId())&&"impala".equals(c.getJdbcName())){
                    return new SimpleDataSource(c.getUrl(),c.getUsername(),c.getPassword(),c.getDriverclassname());
                }
            }
            return null;
        } catch (Exception e) {
            log.error("【动态根据目录获取对应的Impala连接】异常: {}", e.getMessage(), e);
            throw new RuntimeException("【动态根据目录获取对应的Impala连接】异常: " + e.getMessage(), e);
        }
    }

    /**
     * @Method publishPeriodFile
     * @Param
     * @param rootAbsolutePath
     * @param localFilePath
     * @param session
     * @param scriptPublishList
     * @Return void
     * @Description 发布周期脚本文件
     * @Author Wang zhihao
     * @Date 2023/12/1 14:27
     * @Version V1.0
     */
    private void publishPeriodFile(String rootAbsolutePath, String localFilePath, Session session, ScriptPublishListEntity scriptPublishList, String user) {
        scriptPublishList.setState("PUBLISHING");
        // 获取该文件名
        String fileName = FileUtil.getName(rootAbsolutePath);
        // 获取该文件所属目录
        String rootAbsoluteDir = StrUtil.removeAll(rootAbsolutePath, fileName);
        addPublishResultInfo(scriptPublishList.getId(), fileName, "");
        // 若文件已存在则备份
        if (scpTransferService.isExistsPath(session, rootAbsolutePath)) {
            String backupFileName = FileNameUtil.getPrefix(fileName) + "_" + scriptPublishList.getDateFlag() + "." + FileNameUtil.getSuffix(fileName);
            String backupPath = backupFilePath + rootAbsoluteDir;
            if (!StrUtil.equalsIgnoreCase(user, "root")) {
                backupPath = StrUtil.format(USER_ROOT_DIR, user, userBackupPath) + rootAbsoluteDir;
            }
            // 设定备份路径地址
            scriptPublishList.setBackupFilePath(backupPath);
            log.info(StrUtil.format("文件存在，进行备份操作，文件{},将从{}备份到{}",backupFileName, rootAbsolutePath, backupPath));
            scpTransferService.moveFile(session, rootAbsolutePath, backupPath, backupFileName);
            log.info(StrUtil.format("文件{}备份成功", backupFileName));
            scriptPublishList.setBackupFlag("Y");
        }
        // 文件上传
        if (!scpTransferService.isExistsDir(session, rootAbsoluteDir)) {
            log.info(StrUtil.format("目录{}不存在，进行创建操作", rootAbsoluteDir));
            scpTransferService.createDir(session, rootAbsoluteDir);
            log.info(StrUtil.format("目录{}创建成功", rootAbsoluteDir));
        }
        log.info(StrUtil.format("文件{}上传开始", fileName));
        scpTransferService.uploadFile(session, localFilePath, rootAbsolutePath, fileName);
        log.info(StrUtil.format("文件{}上传成功", fileName));
        scriptPublishListService.updateById(scriptPublishList.setId(scriptPublishList.getId()).setState("PUBLISHED"));
    }

    /**
     * @Method executeSql
     * @Param
     * @param dataSource
     * @param sqls
     * @Return void
     * @Description 批量执行SQL
     * @Author Wang zhihao
     * @Date 2023/10/18 10:25
     * @Version V1.0
     */
    private void executeSql(DataSource dataSource, List<String> sqls) {
        Connection connection = null;
        Statement statement = null;
        try {
            connection = dataSource.getConnection();
            statement = connection.createStatement();
            for (String sql : sqls) {
                statement.execute(sql);
            }
        } catch (Exception e) {
            log.error("【执行SQL语句】异常 {}", e.getMessage(), e);
            throw new RuntimeException("【执行SQL语句】异常" + e.getMessage(), e);
        } finally {
            try {
                if (connection != null) {
                    connection.close();
                }
                if (statement != null) {
                    statement.close();
                }
            } catch (Exception e) {
                e.printStackTrace();
                log.error("【连接关闭】异常 {}", e.getMessage(), e);
                throw new RuntimeException("【连接关闭】异常" + e.getMessage(), e);
            }
        }
    }

    /**
     * @Method isImpalaTableExist
     * @Param
     * @param dataSource
     * @param dbName
     * @param tableName
     * @Return boolean
     * @Description 判断impala表是否存在
     * @Author Wang zhihao
     * @Date 2024/1/23 11:10
     * @Version V1.0
     */
    private boolean isImpalaTableExist(DataSource dataSource, String dbName, String tableName) {
        Connection connection = null;
        Statement statement = null;
        ResultSet resultSet = null;
        try {
            connection = dataSource.getConnection();
            statement = connection.createStatement();
            String template = "SHOW TABLES IN {} LIKE '{}'";
            resultSet = statement.executeQuery(StrUtil.format(template, dbName, tableName));
            if (!resultSet.next()) {
                return false;
            }
            return true;
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        } finally {
            DataSourceUtil.close(connection, statement, resultSet);
        }
    }

    /**
     * @Method isOracleTableExist
     * @Param
     * @param dataSource
     * @param tableName
     * @Return boolean
     * @Description 判断Oracle表是否存在
     * @Author Wang zhihao
     * @Date 2024/1/23 11:10
     * @Version V1.0
     */
    private boolean isOracleTableExist(DataSource dataSource, String tableName) {
        Connection connection = null;
        ResultSet resultSet = null;
        try {
            connection = dataSource.getConnection();
            DatabaseMetaData databaseMetaData = connection.getMetaData();
            resultSet = databaseMetaData.getTables(null, null, tableName.toUpperCase(), null);
            return resultSet.next();
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        } finally {
            DataSourceUtil.close(connection, resultSet);
        }
    }

}
